package com.cjm.nio.demo;

import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

public class SubReactorThreadGroup {
    private static final AtomicInteger requestCount = new AtomicInteger(0);
    private final int ioThreadCount;
    private final int businessThreadCount;
    private static final int DEFAULT_NIO_THREAD_COUNT;
    private SubReactorThread[] ioThreads;
    private ExecutorService businessExecutePool;

    static {
        DEFAULT_NIO_THREAD_COUNT = 4;
    }

    public SubReactorThreadGroup() {
        this(DEFAULT_NIO_THREAD_COUNT);
    }

    public SubReactorThreadGroup(int ioThreadCount) {
        if (ioThreadCount < 1) {
            ioThreadCount = DEFAULT_NIO_THREAD_COUNT;
        }
        businessThreadCount = 10;
        businessExecutePool = Executors.newFixedThreadPool(businessThreadCount, new ThreadFactory() {
            private AtomicInteger num = new AtomicInteger(0);
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setName("bussiness-thread-" + num.incrementAndGet());
                return t;
            }
        });
        this.ioThreadCount = ioThreadCount;
        this.ioThreads = new SubReactorThread[ioThreadCount];
        for(int i = 0;i < ioThreadCount;i++) {
            this.ioThreads[i] = new SubReactorThread(businessExecutePool);
            this.ioThreads[i].start();
        }
        System.err.println("nio 线程数量 ： " + ioThreadCount);
    }

    public void dispatch(SocketChannel clientChannel) {
        if (clientChannel != null) {
            next().register(new NioTask(clientChannel, SelectionKey.OP_READ));
        }
    }

    protected SubReactorThread next() {
        return this.ioThreads[requestCount.getAndIncrement() % ioThreadCount];
    }
}
